package com.atguigu.sparkstreaming.output

import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
import org.apache.spark.streaming.kafka010.KafkaUtils
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import org.apache.spark.streaming.{Seconds, StreamingContext}
import redis.clients.jedis.Jedis

/**
 *
 *  优雅地关闭：
 *            在需要程序停止时，让程序自己停止，而不是手动kill!
 *
 *  停止：
 *          StreamingContext.stop()
 *
 *   新启动一个线程，在新线程中进行停止！当前main()在运行到 streamingContext.awaitTermination()就会阻塞！
 *
 *
 */
object StopGracefullyDemo {

  val appName ="StopGracefullyDemo"

  val streamingContext = new StreamingContext("local[*]",appName , Seconds(5))

  def main(args: Array[String]): Unit = {

    val kafkaParams = Map[String, Object](
      "bootstrap.servers" -> "hadoop102:9092,hadoop103:9092",
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> "220309",
      "auto.offset.reset" -> "latest",
      "enable.auto.commit" -> "true"
    )


    val topics = Array("topicA")

    val ds: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream[String, String](
      streamingContext,
      PreferConsistent,
      Subscribe[String, String](topics, kafkaParams)
    )



    //(String, Int):(word,n)
    val ds2: DStream[(String, Int)] = ds.flatMap(record => record.value().split(" "))
      .map((_, 1)).reduceByKey(_+_)

     ds2.print(1000)

    // 启动APP
    streamingContext.start()

    // 在awaitTermination之前运行线程
    new Thread(new MonitorThread).start()

    // 阻塞当前线程，让App一直运行
    streamingContext.awaitTermination()


  }

}

class MonitorThread extends Runnable {

  val jedis = new Jedis("hadoop102", 6379)

  override def run(): Unit = {

    //判断是否应该停止，如果是就停止，如果不是，继续判断
    /*
        如果知道该停止?
            涉及信息的交换(介质，可以是数据库也可以是文件系统)！
            在redis中放一个K-V，当读到V是true时就停止！
     */
    while(true){

      val res: String = jedis.get(StopGracefullyDemo.appName)

      if (("true").equals(res)){

        //关闭逻辑
        StopGracefullyDemo.streamingContext.stop(true,true)

        //退出JVM
        System.exit(0)


      }else{

        //歇会，
        Thread.sleep(5000)

      }

    }
  }
}
